package com.sunday.common.mq.kafka.brave.core;

import brave.kafka.clients.KafkaTracing;
import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.lang.Nullable;

import java.util.Map;
import java.util.Properties;

/**
 * Kafka Consumer 埋点
 * 执行过程:
 * {@link org.springframework.kafka.core.DefaultKafkaConsumerFactory#createConsumer(java.lang.String, java.lang.String, java.lang.String, java.util.Properties) }
 * {@link org.springframework.kafka.listener.KafkaMessageListenerContainer.ListenerConsumer#ListenerConsumer }
 * {@link org.springframework.kafka.config.KafkaListenerEndpointRegistry#start }
 *
 * @author zsy
 * @see brave.kafka.clients.TracingConsumer#TracingConsumer(Consumer, KafkaTracing) 参考
 * @since 2023/11/17
 */
public class TracingKafkaConsumerFactory<K, V> extends DefaultKafkaConsumerFactory<K, V> {

    private final KafkaTracing kafkaTracing;

    public TracingKafkaConsumerFactory(Map<String, Object> configs, KafkaTracing kafkaTracing) {
        super(configs);
        this.kafkaTracing = kafkaTracing;
    }

    @Override
    public Consumer<K, V> createConsumer(@Nullable String groupId, @Nullable String clientIdPrefix,
                                         @Nullable String clientIdSuffix) {
        return kafkaTracing.consumer(super.createConsumer(groupId, clientIdPrefix, clientIdSuffix));
    }

    @Override
    public Consumer<K, V> createConsumer(@Nullable String groupId, @Nullable String clientIdPrefix,
                                         @Nullable final String clientIdSuffixArg, @Nullable Properties properties) {
        return kafkaTracing.consumer(super.createKafkaConsumer(groupId, clientIdPrefix, clientIdSuffixArg, properties));
    }

}
